use crate::cache::get_mysql_pool;
use crate::keys::{BoundTypeKeyTrait, IndexTypeKeyTrait, UniqueTypeKeyTrait, VddKey};
use crate::record_db::RecordDbTrait;
use crate::sqlx::get_sql;
use chrono::{DateTime, Local, NaiveDateTime};
use futures::TryStreamExt;
use serde_json::json;
use sqlx::mysql::MySqlRow;
use sqlx::{Column, Connection, Executor, MySqlPool, Row};
use std::collections::HashMap;
use std::sync::Arc;
use visit_dd_core::field::{VddField, VddFieldRef, VddFieldType};
use visit_dd_core::i18n_text::E_WRONG_KEY_TYPE;
use visit_dd_core::model::{DataStore, DataStoreEngine, VddModel, VddSchema};
use visit_dd_core::proj::VddUser;
use visit_dd_core::query::{
    QueryDataPage, QueryFilter, QueryPlan, ShareQueryParams, WebQueryParams,
};
use visit_dd_core::record::{VddDisplay, VddRecord};
use visit_dd_core::value::VddValue;
use visit_dd_core::visit::ShareRuleOfField;
use visit_dd_core::{i18n, Ares, Astr, FieldId, UserId, DT_FORMAT, EMPTY_FID};
use visit_dd_core::{to_sha256_hash, SqlAttribute};

use super::SqlxDefineTrait;

impl SqlxDefineTrait for MySqlPool {
    async fn update_entity(&self, id: u64, body: &str) -> Result<(), Astr> {
        let r = sqlx::query(&get_sql("update_entity", DataStoreEngine::Mysql))
            .bind(body)
            .bind(id)
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn load_entity(&self, id: u64) -> Result<Astr, Astr> {
        let sql = "SELECT entity_content FROM t_entity WHERE entity_id = ?";
        let r = sqlx::query(sql).bind(id).fetch_one(self).await.unwrap();
        let vu8 = r.get::<Vec<u8>, &str>("entity_content");
        Ok(String::from_utf8(vu8).unwrap().into())
    }

    async fn load_entities(&self, vk: VddKey, deleted: bool) -> Result<Vec<Astr>, Astr> {
        if !vk.entity_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = "SELECT entity_content FROM t_entity WHERE entity_type = ? AND deleted = ?";
        let rows = sqlx::query(&sql)
            .bind(Into::<u16>::into(&vk) as i16)
            .bind(if deleted { 1 } else { 0 })
            .fetch_all(self)
            .await
            .unwrap();
        let mut r_vec = Vec::<Astr>::new();
        rows.iter().for_each(|row| {
            let vu8 = row.get::<Vec<u8>, &str>("entity_content");
            r_vec.push(String::from_utf8(vu8).unwrap().into());
        });
        Ok(r_vec)
    }

    async fn load_entities_form_index_query(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        deleted: bool,
    ) -> Result<Vec<Astr>, Astr> {
        let sql = if vk.get_index_key_as_u64() > 0 {
            "SELECT entity_content FROM t_entity WHERE entity_id IN (SELECT index_item FROM t_index WHERE index_type = ? AND arg_u64 = ?) AND deleted = ?"
        } else {
            "SELECT entity_content FROM t_entity WHERE entity_id IN (SELECT index_item FROM t_index WHERE index_type = ?) AND deleted = ?"
        };
        let rows = if vk.get_index_key_as_u64() > 0 {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(vk.get_index_key_as_u64())
                .bind(if deleted { 1 } else { 0 })
                .fetch_all(self)
                .await
                .unwrap()
        } else {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(if deleted { 1 } else { 0 })
                .fetch_all(self)
                .await
                .unwrap()
        };
        let mut r_vec = Vec::<Astr>::new();
        rows.iter().for_each(|row| {
            let vu8 = row.get::<Vec<u8>, &str>("entity_content");
            r_vec.push(String::from_utf8(vu8).unwrap().into());
        });
        Ok(r_vec)
    }

    async fn count_entities(&self, vk: VddKey) -> Result<usize, Astr> {
        if !vk.entity_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = "SELECT count(*) AS count FROM t_entity WHERE entity_type = ? AND deleted = 0";
        let row = sqlx::query(&sql)
            .bind(Into::<u16>::into(&vk))
            .fetch_one(self)
            .await
            .unwrap();
        let r = row.get::<i64, &str>("count");
        Ok(r as usize)
    }

    async fn delete_entity(&self, id: u64) -> Result<(), Astr> {
        let sql = get_sql("delete_entity", DataStoreEngine::Mysql);
        let r = sqlx::query(&sql).bind(id).execute(self).await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_entities_from_index_query(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Result<(), Astr> {
        let sql = "UPDATE t_entity SET deleted = 1 WHERE entity_id IN (SELECT index_item FROM t_index WHERE index_type = ? AND arg_u64 = ?)";
        let r = sqlx::query(sql)
            .bind(Into::<u16>::into(vk.as_ref()))
            .bind(vk.get_index_key_as_u64())
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn add_to_index(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        index_item: u64,
    ) -> Result<(), Astr> {
        let sql = &get_sql("insert_index", DataStoreEngine::Mysql);
        let r = sqlx::query(&sql)
            .bind(Into::<u16>::into(vk.as_ref()))
            .bind(index_item)
            .bind(vk.get_index_key_as_u64())
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_index(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        index_item: u64,
    ) -> Result<(), Astr> {
        let sql = "DELETE FROM t_index WHERE index_type = ? AND index_item = ? AND arg_u64 = ?";
        let r = sqlx::query(&sql)
            .bind(Into::<u16>::into(vk.as_ref()))
            .bind(index_item)
            .bind(vk.get_index_key_as_u64())
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_indexes(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Result<(), Astr> {
        let sql = "DELETE FROM t_index WHERE index_type = ? AND arg_u64 = ?";
        let r = sqlx::query(&sql)
            .bind(Into::<u16>::into(vk.as_ref()))
            .bind(vk.get_index_key_as_u64())
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn read_index(&self, vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>) -> Vec<i64> {
        let sql = if vk.get_index_key_as_u64() > 0 {
            "SELECT index_item FROM t_index WHERE index_type = ? AND arg_u64 = ? ORDER BY index_item"
        } else {
            "SELECT index_item FROM t_index WHERE index_type = ? ORDER BY index_item"
        };
        let rows = if vk.get_index_key_as_u64() > 0 {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(vk.get_index_key_as_u64())
                .fetch_all(self)
                .await
        } else {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .fetch_all(self)
                .await
        };
        let mut r_vec = Vec::<i64>::new();
        rows.unwrap().iter().for_each(|row| {
            let item = row.get::<i64, &str>("index_item");
            r_vec.push(item);
        });
        r_vec
    }

    async fn count_in_index(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Result<usize, Astr> {
        let sql = "SELECT COUNT(*) AS count FROM t_index WHERE index_type = ? AND arg_u64 = ?";
        let row = sqlx::query(&sql)
            .bind(Into::<u16>::into(vk.as_ref()))
            .bind(vk.get_index_key_as_u64())
            .fetch_one(self)
            .await
            .unwrap();
        let r = row.get::<i64, &str>("count");
        Ok(r as usize)
    }

    async fn get_unique(
        &self,
        vk_u: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Result<u64, Astr> {
        let sql = "SELECT link_id FROM t_unique WHERE unique_type = ? AND hash = ? AND arg_u64 = ?";
        let row = sqlx::query(sql)
            .bind(Into::<u16>::into(vk_u.as_ref()))
            .bind(vk_u.get_unique_hash().unwrap().to_string())
            .bind(vk_u.get_unique_arg())
            .fetch_optional(self)
            .await
            .unwrap();
        if row.is_none() {
            Err("record not found".into())
        } else {
            let r = row.unwrap().get::<i64, _>("link_id") as u64;
            Ok(r)
        }
    }

    async fn delete_unique_by_value(
        &self,
        vk: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        link_id: u64,
    ) -> Result<(), Astr> {
        if !vk.unique_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = "DELETE FROM t_unique WHERE unique_type = ? AND link_id = ? AND arg_u64 = ?";
        let r = sqlx::query(sql)
            .bind(Into::<u16>::into(vk.as_ref()))
            .bind(link_id)
            .bind(vk.get_unique_arg())
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_uniques_from_index_query(
        &self,
        vk_u: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        vk_i: VddKey,
    ) -> Result<(), Astr> {
        if !vk_u.unique_type() || !vk_i.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = "DELETE FROM t_unique WHERE link_id IN (SELECT index_item FROM t_index WHERE index_type = ? AND arg_u64 = ?) AND unique_type = ?";
        let r = sqlx::query(sql)
            .bind(Into::<u16>::into(&vk_i))
            .bind(vk_i.get_index_key_as_u64())
            .bind(Into::<u16>::into(vk_u.as_ref()))
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn select_uniques(
        &self,
        vk_u: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> HashMap<String, u64> {
        if !vk_u.unique_type() {
            panic!("{}", i18n!(E_WRONG_KEY_TYPE));
        }
        let hash = vk_u.get_unique_hash();
        let sql = if hash.is_ok() {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = ? AND hash = ? AND arg_u64 = ?"
        } else {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = ? AND arg_u64 = ?"
        };
        let rows = if hash.is_ok() {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk_u.as_ref()))
                .bind(hash.unwrap().as_ref())
                .bind(vk_u.get_unique_arg())
                .fetch_all(self)
                .await
                .unwrap()
        } else {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk_u.as_ref()))
                .bind(vk_u.get_unique_arg())
                .fetch_all(self)
                .await
                .unwrap()
        };
        let mut r = HashMap::<String, u64>::new();
        rows.iter().for_each(|row| {
            r.insert(
                String::from_utf8(row.get::<Vec<u8>, &str>("hash")).unwrap(),
                row.get::<i64, &str>("link_id") as u64,
            );
        });
        r
    }

    async fn select_uniques_by_value(
        &self,
        vk: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        link_id: Option<u64>,
    ) -> HashMap<String, u64> {
        if !vk.unique_type() {
            panic!("{}", i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = if link_id.is_some() {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = ? AND link_id = ?"
        } else {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = ?"
        };
        let rows = if link_id.is_some() {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(link_id.unwrap())
                .fetch_all(self)
                .await
                .unwrap()
        } else {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .fetch_all(self)
                .await
                .unwrap()
        };
        let mut r = HashMap::<String, u64>::new();
        rows.iter().for_each(|row| {
            r.insert(
                String::from_utf8(row.get::<Vec<u8>, &str>("hash")).unwrap(),
                row.get::<i64, &str>("link_id") as u64,
            );
        });
        r
    }

    async fn select_uniques_from_index_query(
        &self,
        vk_u: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        vk_i: VddKey,
    ) -> HashMap<String, u64> {
        if !vk_u.unique_type() || !vk_i.index_type() {
            panic!("{}", i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = "SELECT hash,link_id FROM t_unique WHERE unique_type = ? AND link_id IN (SELECT index_item FROM t_index WHERE index_type = ? AND arg_u64 = ?)";
        let rows = sqlx::query(sql)
            .bind(Into::<u16>::into(vk_u.as_ref()))
            .bind(Into::<u16>::into(&vk_i))
            .bind(vk_i.get_index_key_as_u64())
            .fetch_all(self)
            .await
            .unwrap();
        let mut r = HashMap::<String, u64>::new();
        rows.iter().for_each(|row| {
            r.insert(
                String::from_utf8(row.get::<Vec<u8>, &str>("hash")).unwrap(),
                row.get::<i64, &str>("link_id") as u64,
            );
        });
        r
    }

    async fn save_bound(
        &self,
        vk: impl BoundTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        bound_content: &str,
    ) -> Result<(), Astr> {
        if !vk.bound_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        sqlx::query(&get_sql("delete_bound", DataStoreEngine::Mysql))
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(vk.get_bound_left() as i64)
            .bind(vk.get_bound_right_as_u64().unwrap() as i64)
            .execute(self)
            .await
            .unwrap();
        sqlx::query(&get_sql("insert_bound", DataStoreEngine::Mysql))
            .bind(Into::<u16>::into(vk.as_ref()))
            .bind(vk.get_bound_left())
            .bind(vk.get_bound_right_as_u64().unwrap())
            .bind(bound_content)
            .execute(self)
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    /// 多个绑定，根据绑定类型和绑定左值，绑定右值可选
    async fn select_bounds(
        &self,
        vk: impl BoundTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Vec<String> {
        let bound_left = vk.get_bound_left();
        let use_bound_left = bound_left > 0;
        let use_bound_right =
            vk.get_bound_right_as_u64().is_ok() && vk.get_bound_right_as_u64().unwrap() > 0;
        let mut sql = "SELECT bound_content FROM t_bound WHERE bound_type = ?".to_string();
        if use_bound_left {
            sql = format!("{} AND bound_left = ?", sql);
        }
        if use_bound_right {
            sql = format!("{} AND bound_right = ?", sql);
        }
        let rows = match (use_bound_left, use_bound_right) {
            (false, false) => sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .fetch_all(self)
                .await
                .unwrap(),
            (true, false) => sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(bound_left)
                .fetch_all(self)
                .await
                .unwrap(),
            (true, true) => sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(bound_left)
                .bind(vk.get_bound_right_as_u64().unwrap())
                .fetch_all(self)
                .await
                .unwrap(),
            (false, true) => sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(vk.get_bound_right_as_u64().unwrap())
                .fetch_all(self)
                .await
                .unwrap(),
        };
        rows.iter()
            .map(|row| row.get::<Vec<u8>, &str>("bound_content"))
            .map(|x| String::from_utf8(x).unwrap())
            .collect::<Vec<String>>()
    }

    /// 删除多个绑定，根据绑定类型和绑定左值，绑定右值可选
    async fn delete_bounds(
        &self,
        vk: impl BoundTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Result<(), Astr> {
        let mut sql = "DELETE FROM t_bound WHERE bound_type = ? AND bound_left = ?".to_string();
        let mut flag_1 = false;
        match vk.get_bound_right_as_u64() {
            Ok(right) => {
                if right > 0 {
                    sql.push_str(" AND bound_right = ?");
                    flag_1 = true
                }
            }
            Err(_) => {}
        }
        let r = if flag_1 {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(vk.get_bound_left())
                .bind(vk.get_bound_right_as_u64().unwrap())
                .execute(self)
                .await
        } else {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()))
                .bind(vk.get_bound_left())
                .execute(self)
                .await
        };
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn count_bounds_grouped(&self, vk_s: &[VddKey]) -> HashMap<u64, usize> {
        if vk_s.len() < 1 {
            return HashMap::new();
        }
        let mut sql_part = String::new();
        vk_s.iter().for_each(|bound_type| {
            sql_part.push_str(&format!(",{}", bound_type.get_bound_left()));
        });
        if vk_s.len() > 0 {
            sql_part.remove(0);
        }
        let sql = format!("SELECT bound_left,count(*) AS c FROM t_bound WHERE bound_type = {} GROUP BY bound_left HAVING bound_left IN ({})", Into::<u16>::into(&vk_s[0]), sql_part);
        let rows = sqlx::query(&sql).fetch_all(self).await.unwrap();
        let mut r = HashMap::<u64, usize>::new();
        rows.iter().for_each(|row| {
            let key = row.get::<i64, _>("bound_left");
            let value = row.get::<i64, _>("c");
            r.insert(key as u64, value as usize);
        });
        r
    }

    /// 需要事务的函数集
    async fn u_register(&self, user: &VddUser) -> Result<(), Astr> {
        let mut tx = self.begin().await.unwrap();
        let sql = get_sql("insert_entity", DataStoreEngine::Mysql);
        tracing::info!(sql);
        tx.execute(
            sqlx::query(&sql)
                .bind(user.id.0)
                .bind(Into::<u16>::into(&VddKey::EntityAppUser(UserId(0))))
                .bind(serde_json::to_string(user).unwrap()),
        )
        .await
        .unwrap();
        let vk_u = VddKey::UniquePhone(user.phone.clone());
        tx.execute(
            sqlx::query(&get_sql("insert_unique", DataStoreEngine::Mysql))
                .bind(Into::<u16>::into(&vk_u))
                .bind(user.phone.to_string())
                .bind(user.id.0)
                .bind(0),
        )
        .await
        .unwrap();
        let r = tx.commit().await;
        r.map_err(|e| e.to_string().into())
    }

    async fn m_save_new(&self, model: &VddModel) -> Result<(), Astr> {
        let mut tx = self.begin().await.unwrap();
        let sql = get_sql("insert_entity", DataStoreEngine::Mysql);
        tx.execute(
            sqlx::query(&sql)
                .bind(model.id)
                .bind(Into::<u16>::into(&VddKey::EntityModel(0)))
                .bind(&serde_json::to_string(model).unwrap()),
        )
        .await
        .unwrap();
        let sql = get_sql("insert_index", DataStoreEngine::Mysql);
        let it = VddKey::IndexUser2Model(model.uid.clone());
        tx.execute(
            sqlx::query(&sql)
                .bind(Into::<u16>::into(&it))
                .bind(model.id)
                .bind(it.get_index_key_as_u64()),
        )
        .await
        .unwrap();
        let r = tx.commit().await;
        r.map_err(|e| e.to_string().into())
    }

    async fn m_save_tags(
        &self,
        model: &VddModel,
        model_tag: &str,
        tag_map: &HashMap<Astr, u64>,
    ) -> Result<(), Astr> {
        // 先清空旧的标记
        let vk_u = VddKey::UniqueModelTag(model_tag.into());
        self.delete_unique_by_value(vk_u.clone(), model.id).await?;
        // 检测hash
        if self.get_unique(vk_u.clone()).await.is_ok() {
            return Err("已有同名模型标记".into());
        }
        for x in tag_map.iter() {
            let vk_u2 = VddKey::UniqueFieldTag(model_tag.into(), x.0.clone());
            self.delete_unique_by_value(vk_u2.clone(), *x.1).await?;
        }
        // 再插入新的标记
        let mut tx = self.begin().await.unwrap();
        tx.execute(
            sqlx::query(&get_sql("insert_unique", DataStoreEngine::Mysql))
                .bind(Into::<u16>::into(&vk_u))
                .bind(model_tag)
                .bind(model.id)
                .bind(0u64),
        )
        .await
        .unwrap();
        for x in tag_map.iter() {
            let field_tag = format!("{}:{}", model_tag, x.0);
            let vk_u2 = VddKey::UniqueFieldTag(model_tag.into(), x.0.clone());
            tx.execute(
                sqlx::query(&get_sql("insert_unique", DataStoreEngine::Mysql))
                    .bind(Into::<u16>::into(&vk_u2))
                    .bind(field_tag)
                    .bind(*x.1)
                    .bind(model.id),
            )
            .await
            .unwrap();
        }
        tx.commit().await.map_err(|e| e.to_string().into())
    }

    async fn f_save_new(&self, field: &VddField) -> Ares {
        let mut tx = self.begin().await.unwrap();
        tx.execute(
            sqlx::query(&get_sql("insert_entity", DataStoreEngine::Mysql))
                .bind(field.id.0)
                .bind(Into::<u16>::into(&VddKey::EntityField(EMPTY_FID)))
                .bind(serde_json::to_string(field).unwrap()),
        )
        .await
        .map_err(|e| Arc::from(e.to_string()))?;
        let it = VddKey::IndexModel2Field(field.model_id);
        tx.execute(
            sqlx::query(&get_sql("insert_index", DataStoreEngine::Mysql))
                .bind(Into::<u16>::into(&it))
                .bind(field.id.0)
                .bind(it.get_index_key_as_u64()),
        )
        .await
        .map_err(|e| Arc::from(e.to_string()))?;
        tx.commit().await.map_err(|e| e.to_string().into())
    }

    async fn f_delete(&self, model_id: u64, field_id: FieldId) -> Ares {
        let mut tx = self.begin().await.unwrap();
        let sql = get_sql("delete_index", DataStoreEngine::Mysql);
        let it = VddKey::IndexModel2Field(model_id);
        tx.execute(
            sqlx::query(&sql)
                .bind(Into::<u16>::into(&it))
                .bind(field_id.0)
                .bind(it.get_index_key_as_u64()),
        )
        .await
        .unwrap();
        let sql = get_sql("delete_entity", DataStoreEngine::Mysql);
        tx.execute(sqlx::query(&sql).bind(field_id.0))
            .await
            .unwrap();
        tx.commit().await.map_err(|e| e.to_string().into())
    }

    async fn m_clear_tags(&self, model_id: u64, field_ids: &[FieldId]) -> Result<(), Astr> {
        let mut tx = self.begin().await.unwrap();
        tx.execute(
            sqlx::query(&get_sql("delete_unique", DataStoreEngine::Mysql))
                .bind(Into::<u16>::into(&VddKey::UniqueModelTag("".into())))
                .bind(model_id),
        )
        .await
        .unwrap();
        for x in field_ids {
            tx.execute(
                sqlx::query(&get_sql("delete_unique", DataStoreEngine::Mysql))
                    .bind(Into::<u16>::into(&VddKey::UniqueFieldTag(
                        "".into(),
                        "".into(),
                    )))
                    .bind(x.0),
            )
            .await
            .unwrap();
        }
        tx.commit().await.map_err(|e| e.to_string().into())
    }

    async fn do_delete_entity(&self, id: u64) -> Result<(), Astr> {
        sqlx::query(&get_sql("do_delete_entity", DataStoreEngine::Mysql))
            .bind(id as i64)
            .execute(self)
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    async fn restore_entity(&self, id: u64) -> Ares {
        sqlx::query(&get_sql("restore_entity", DataStoreEngine::Mysql))
            .bind(id as i64)
            .execute(self)
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    async fn check_entity_deleted(&self, id: u64) -> bool {
        let sql_r = sqlx::query(&get_sql("check_entity_deleted", DataStoreEngine::Mysql))
            .bind(id as i64)
            .fetch_one(self)
            .await
            .unwrap();
        sql_r.get::<i64, &str>("count") as usize > 0
    }

    async fn save_plan(&self, plan: &QueryPlan, is_update: bool) -> Ares {
        let hash = to_sha256_hash(plan.name.as_bytes());
        let vk_u = VddKey::UniqueQueryPlan(plan.u_id.clone(), hash.clone());
        self.delete_unique_by_value(vk_u.clone(), plan.id).await?;
        let vk_b = VddKey::BoundQueryPlan(plan.u_id.clone(), plan.id);
        self.delete_bounds(vk_b.clone()).await?;
        let id2 = self.get_unique(vk_u.clone()).await;
        if id2.is_ok() && id2.unwrap() == plan.id && !is_update {
            return Err("已有同名称查询计划存在".into());
        }
        let mut tx = self.begin().await.unwrap();
        tx.execute(
            sqlx::query(&get_sql("insert_bound", DataStoreEngine::Mysql))
                .bind(Into::<u16>::into(&vk_b))
                .bind(vk_b.get_bound_left())
                .bind(vk_b.get_bound_right_as_u64().unwrap())
                .bind(json!(plan).to_string()),
        )
        .await
        .unwrap();
        tx.execute(
            sqlx::query(&get_sql("insert_unique", DataStoreEngine::Mysql))
                .bind(Into::<u16>::into(&vk_u))
                .bind(hash.to_string())
                .bind(plan.id)
                .bind(plan.u_id.0),
        )
        .await
        .unwrap();
        tx.commit().await.map_err(|e| e.to_string().into())
    }
}

pub(crate) async fn test_connection(ds: &DataStore) -> bool {
    get_mysql_pool(ds)
        .await
        .acquire()
        .await
        .unwrap()
        .ping()
        .await
        .is_ok()
}

/// 适配mysql的数据类型
fn judge_field_type(s: &str) -> VddFieldType {
    tracing::info!("judge field type:{}", s);
    if s.starts_with("int") {
        VddFieldType::FtInt {
            min_v: None,
            max_v: None,
            unit_label: None,
        }
    } else if s.eq_ignore_ascii_case("timestamp") || s.eq_ignore_ascii_case("datetime") {
        VddFieldType::FtDateTime("".into())
    } else {
        VddFieldType::FtString {
            min_len: None,
            max_len: None,
            pattern: None,
        }
    }
}

fn row_to_record(row: &MySqlRow, fields: &[VddFieldRef], record: &mut VddRecord) {
    fn _extract_col_ts(row: &MySqlRow, col_name: &str) -> Arc<VddValue> {
        match row.try_get::<DateTime<Local>, &str>(col_name) {
            Err(e) => Arc::new(VddValue::S101(e.to_string())),
            Ok(t) => Arc::new(VddValue::S101(t.format(DT_FORMAT).to_string())),
        }
    }
    fn _extract_col_dt(row: &MySqlRow, col_name: &str) -> Arc<VddValue> {
        match row.try_get::<NaiveDateTime, &str>(col_name) {
            Err(e) => Arc::new(VddValue::S101(e.to_string())),
            Ok(t) => Arc::new(VddValue::S101(t.format(DT_FORMAT).to_string())),
        }
    }
    row.columns().iter().for_each(|col| {
        let col_name = col.name();
        let field = fields
            .iter()
            .filter(|f| f.label == col_name)
            .next()
            .unwrap();
        let col_type = col.type_info().to_string();
        println!("\nmysql col type: {:?}", col_type);
        let v = match col_type.as_str() {
            "TIMESTAMP" => _extract_col_ts(&row, &col_name),
            "INT UNSIGNED" | "BIGINT UNSIGNED" => match row.try_get::<u64, &str>(col_name) {
                Err(e) => Arc::new(VddValue::S101(e.to_string())),
                Ok(t) => Arc::new(VddValue::S201(t as i64)),
            },
            "INT" | "BIGINT" => match row.try_get::<i64, &str>(col_name) {
                Err(e) => Arc::new(VddValue::S101(e.to_string())),
                Ok(t) => Arc::new(VddValue::S201(t)),
            },
            _ => match row.try_get::<&str, &str>(col_name) {
                Err(e) => Arc::new(VddValue::S101(e.to_string())),
                Ok(t) => Arc::new(VddValue::S101(t.to_string())),
            },
        };
        record.value.insert(field.id.clone(), v);
        // if col_type.eq_ignore_ascii_case("datetime") {
        //     record
        //         .value
        //         .insert(field.id.clone(), _extract_col_dt(&row, &col_name));
        // }
        // if col_type.eq_ignore_ascii_case("float") {
        //     match row.try_get::<f32, &str>(col_name) {
        //         Err(e) => {
        //             record.value.insert(
        //                 field.id.clone(),
        //                 Arc::new(VddValue::new_101(&e.to_string())),
        //             );
        //         }
        //         Ok(t) => {
        //             record
        //                 .value
        //                 .insert(field.id.clone(), Arc::new(VddValue::new_s202(t as f64)));
        //         }
        //     }
        // }
        // if col_type.eq_ignore_ascii_case("double") {
        //     match row.try_get::<f64, &str>(col_name) {
        //         Err(e) => {
        //             record.value.insert(
        //                 field.id.clone(),
        //                 Arc::new(VddValue::new_101(&e.to_string())),
        //             );
        //         }
        //         Ok(t) => {
        //             record
        //                 .value
        //                 .insert(field.id.clone(), Arc::new(VddValue::new_s202(t)));
        //         }
        //     }
        // }
    });
}
///
fn extract_record(row: &MySqlRow, schema: &VddSchema) -> Arc<VddRecord> {
    let mut record = VddRecord::new_default(schema.model.uid.clone(), schema.model.id);
    row_to_record(row, &schema.fields, &mut record);
    Arc::new(record)
}

pub(crate) async fn table_to_fields(model: &VddModel) -> Vec<VddFieldRef> {
    let store = model.store.clone().unwrap();
    let pool = get_mysql_pool(store.as_ref()).await;
    let sql = format!("describe {}", store.remote_key);
    let mut rows = sqlx::query(&sql).fetch(&pool);
    let mut field_vec = Vec::<VddFieldRef>::new();
    let mut display_order = 1;
    while let Some(row) = rows.try_next().await.unwrap() {
        let field_type = row.try_get::<&str, &str>("Type").unwrap();
        let col_name = row.try_get::<&str, &str>("Field").unwrap();
        let field = VddField::new(
            model.id,
            display_order,
            col_name,
            judge_field_type(field_type).get_type_id(),
        );
        field_vec.push(Arc::new(field));
        display_order = display_order + 1;
    }
    field_vec
}

/// 生成查询语句
fn gen_select_sql(schema: &VddSchema, skip_col_names: &[Astr], q: &WebQueryParams) -> String {
    let mut cols = String::new();
    schema.fields.iter().for_each(|field| {
        let col_name = schema.get_field_tag(&field.id);
        // 跳过忽略的列名
        if !skip_col_names.contains(&col_name) {
            cols.push_str(&format!(",`{}`", col_name));
        }
    });
    cols.remove(0);
    let store = schema.model.store.clone().unwrap();
    let mut sql = format!("SELECT {} FROM `{}` WHERE 1 = 1", cols, &store.remote_key);
    // 增加条件过滤支持
    let mut parts = Vec::<String>::new();
    q.filters.iter().for_each(|filter| match filter.as_ref() {
        QueryFilter::QfFiledValue(fve) => {
            let field_id = fve.get_field_id();
            let field = schema.try_get_field(&field_id).unwrap();
            let field_tag = schema.get_field_tag(&field_id);
            let part = fve.to_mysql_where_part(&field_tag, field.quote_value());
            if !part.is_empty() {
                parts.push(part);
            }
        }
        _ => {}
    });
    if !parts.is_empty() {
        sql = format!("{} AND {}", sql, parts.join(" AND "));
    }
    sql
}

pub(crate) async fn get_page_sqlx(
    schema: &VddSchema,
    q: &WebQueryParams,
) -> QueryDataPage<VddDisplay> {
    let store = schema.model.store.clone().unwrap();
    let pk_fid = schema.get_model_remote_primary_key_as_fid().unwrap();
    let pool = get_mysql_pool(store.as_ref()).await;
    let sql = gen_select_sql(schema, &Vec::<Astr>::new(), q);
    tracing::info!("run in mysql with {}", sql);
    let mut rows = sqlx::query(&sql).fetch(&pool);
    let mut records = Vec::<Arc<VddDisplay>>::new();
    let mut last_index = 0u64;
    while let Some(row) = rows.try_next().await.unwrap() {
        let rd = extract_record(&row, &schema);
        last_index = TryInto::<i64>::try_into(rd.get_value(&pk_fid).as_ref()).unwrap() as u64;
        let rdd = rd.new_display(&schema).await;
        records.push(Arc::new(rdd));
        if records.len() >= q.page_size as usize {
            break;
        }
    }
    QueryDataPage {
        data: records,
        all_record_count: 0,
        last_index,
    }
}

/// 模型在共享状态下支持的查询
pub(crate) async fn share_page(sqp: &ShareQueryParams) -> QueryDataPage<VddDisplay> {
    let store = sqp.schema.model.store.clone().unwrap();
    let pool = get_mysql_pool(store.as_ref()).await;
    // 分享限制中的不可见字段转换成列名
    let mut skip_fid = Vec::<FieldId>::new();
    sqp.vg.rule_f.iter().for_each(|x| match x.as_ref() {
        ShareRuleOfField::SrfHide(field_ids) => {
            for fid in field_ids {
                skip_fid.push(fid.clone());
            }
        }
        _ => {}
    });
    let skip_col_names = sqp
        .schema
        .fields
        .iter()
        .map(|field| {
            if skip_fid.contains(&field.id) {
                Some(sqp.schema.get_field_tag(&field.id))
            } else {
                None
            }
        })
        .filter(|x| x.is_some())
        .map(|x| x.unwrap())
        .collect::<Vec<Astr>>();
    let sql = gen_select_sql(&sqp.schema, &skip_col_names, &sqp.params);
    tracing::info!("run in mysql with {}", sql);
    let mut rows = sqlx::query(&sql).fetch(&pool);
    let mut records = Vec::<Arc<VddDisplay>>::new();
    while let Some(row) = rows.try_next().await.unwrap() {
        let record = extract_record(&row, &sqp.schema);
        let recordd = record.new_display(&sqp.schema).await;
        records.push(Arc::new(recordd));
        if records.len() >= sqp.params.page_size as usize {
            break;
        }
    }
    QueryDataPage {
        data: records,
        all_record_count: 0,
        last_index: 0,
    }
}

// pub(crate) async fn try_run_sql(
//     requester: &UserId,
//     schemas: &[Arc<VddSchema>],
//     sql: &str,
// ) -> Result<Arc<SqlQueryResult>, Arc<str>> {
//     fn _extract_fields_form_row(row: &MySqlRow) -> Vec<VddFieldRef> {
//         let mut fields = Vec::<VddFieldRef>::new();
//         let mut display_order = 1;
//         row.columns().iter().for_each(|col| {
//             let col_name = col.name();
//             let col_type = col.type_info().name();
//             let cc: CommonCode = (&_judge_field_type(col_type)).into();
//             let field = VddField::new(0, display_order, col_name, cc.code);
//             fields.push(Arc::new(field));
//             display_order += 1;
//         });
//         fields
//     }
//     let store = schemas.first().unwrap().model.store.clone().unwrap();
//     let real_sql = sqlx_replace_tags(schemas, sql);
//     let pool = get_mysql_pool(&store).await;
//     let mut rows = sqlx::query(&real_sql).fetch(&pool);
//     let mut records = Vec::<VddDisplayRef>::new();
//     let mut fields = Vec::<VddFieldRef>::new();
//     let mut row_index = 0;
//     while let Some(row) = rows.try_next().await.unwrap() {
//         if row_index < 1 {
//             fields = _extract_fields_form_row(&row);
//         }
//         let mut record = VddRecord::new_default(requester, 0);
//         _row_to_record(&row, &fields, &mut record);
//         records.push(Arc::new(VddDisplay::from(Arc::new(record))));
//         row_index += 1;
//     }
//     Ok(Arc::new(SqlQueryResult {
//         fields,
//         records,
//         ts: String::new(),
//     }))
// }

// 向集群注册实例，成功则返回
// pub(crate) async fn register_instance(config: &AppConfig) -> Result<u8, Astr> {
//     let pool = get_mysql_pool(&config.meta_store.clone().unwrap()).await;
//     let sql = "SELECT COUNT(*) AS count FROM t_cluster WHERE cluster_name = ?";
//     let row = sqlx::query(&sql)
//         .bind(config.cluster_name.to_string())
//         .fetch_one(&pool)
//         .await
//         .unwrap();
//     let r = row.get::<i64, &str>("count");
//     if r < 8 {
//         Ok(r as u8)
//     } else {
//         Err("集群已满，不能加入".into())
//     }
// }
